import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;

public class SparkWindowTimeCalc {
    public static void main(String[] args) {
        calcSparkWindow();
    }

    // 计算event-time所属窗口
    public static void calcSparkWindow() {
        long windowDuration = 10;
        long slideDuration = 5;

        windowDuration = windowDuration * 1000 * 1000;
        slideDuration = slideDuration * 1000 * 1000;

        int maxNumOverlapping = (int)Math.ceil(windowDuration / slideDuration);

        String time = "2019-10-12 08:59:55";
//        time = "2019-10-12 09:00:02";
        for (int i = 0; i < maxNumOverlapping; i++) {
            long micro = timeToMicro(time);
            System.out.println("micro : " + micro);
            double devision = (micro - 0)/slideDuration;
            int ceil = (int)Math.ceil(devision);
            int windowId = ceil;
            if (devision == ceil) {
                windowId = ceil + 1;// 能整除就向后推一个滑动窗口
            }
            System.out.println("windowId: " + windowId);
            long start = (windowId + i - maxNumOverlapping) * slideDuration + 0;
            long end = start + windowDuration;
            System.out.println(format(start/1000) + ", " + format(end/1000));
            System.out.println("---------------------");
        }

        /**
         micro : 1570841995000000
         windowId: 314168400
         2019-10-12 08:59:50, 2019-10-12 09:00:00
         ---------------------
         micro : 1570841995000000
         windowId: 314168400
         2019-10-12 08:59:55, 2019-10-12 09:00:05
         ---------------------
         ******************************************************************
         micro : 1570842002000000
         windowId: 314168401
         2019-10-12 08:59:55, 2019-10-12 09:00:05
         ---------------------
         micro : 1570842002000000
         windowId: 314168401
         2019-10-12 09:00:00, 2019-10-12 09:00:10
         */
    }

    public static String format(long time) {
        return LocalDateTime.ofInstant(new Date(time).toInstant(), ZoneId.systemDefault()).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    }
    public static long timeToMicro(String time) {
        return LocalDateTime.parse(time, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli() * 1000;
    }
}
